package org.quxiao.hbase.test.singleAttrQuery;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;

public class HbaseSingleAttrQueryTester implements SingleAttrQueryTester {

	private Configuration conf = null;
	private HBaseAdmin admin = null;
	private String tableName = "";
	private String columnFamily = "";
	private String key = "";
	private String value = ""; 
	
	public HbaseSingleAttrQueryTester (String tableName, String columnFamily, String key, String value)
	{
		this.tableName = tableName;
		this.columnFamily = columnFamily;
		this.key = key;
		this.value = value;
	}
	
	@Override
	public long queryTime() throws SingleAttrQueryException {
		// TODO Auto-generated method stub
		connect();
		Scan scan = new Scan();
		scan.addColumn(columnFamily.getBytes(), key.getBytes());
		Filter filter = 
			new SingleColumnValueFilter(columnFamily.getBytes(), key.getBytes(), CompareFilter.CompareOp.EQUAL, value.getBytes());
		scan.setFilter(filter);
		scan.setCaching(1024*10);
		
		long t = 0;
		int count = 0;
		try
		{
			HTable table = new HTable(conf, tableName);
			table.setAutoFlush(false);
			
			ResultScanner rs = table.getScanner(scan);
			t = System.currentTimeMillis();
			for (Result result: rs)
			{
				++ count;
			}
			t = System.currentTimeMillis() - t;
//			System.out.println(count);
			rs.close();
		}
		catch (IOException ioe)
		{
			System.err.println(ioe);
			throw new SingleAttrQueryException();
		}
		return t;
	}

	private void connect () throws SingleAttrQueryException
	{
		conf = HBaseConfiguration.create();
		try
		{
			admin = new HBaseAdmin(conf);
		}
		catch (ZooKeeperConnectionException zooEx)
		{
			System.err.println("can not connect zookeeper!");
			throw new SingleAttrQueryException();
		}
		catch (MasterNotRunningException masterNotRunEx)
		{
			System.err.println("hbase master isn't running!");
			throw new SingleAttrQueryException();
		}
	}
	
	public String toString ()
	{
		return "HBase single attribute query tester for table " + this.tableName;
	}
	
	/**
	 * @param args
	 */
	public static void main(String[] args) {
		// TODO Auto-generated method stub

	}

}
